package dacp.etl.kafka.hdfs.tools;

import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.UUID;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import io.confluent.connect.hdfs.Format;
import io.confluent.connect.hdfs.HdfsSinkConnectorConfig;
import io.confluent.connect.hdfs.RecordWriterProvider;
import io.confluent.connect.hdfs.storage.Storage;
import io.confluent.connect.hdfs.storage.StorageFactory;
import jline.internal.Log;

public class CosumerToHdfs {
	private static Storage storage;
	private static String url;
	private static final String lineSeparator = "\n";
	private static boolean onlyread = false;
	private static boolean onlywrite = false;
	private static String example = "|712|5|0000000a27d5c3acc25e20ae597d37ff|6|460025071280390|3594540725330202|18327663707|20|1501378581109|1501378581118|0|20|65535|0|255|255|255|1509965509|65535|255|4294967295|807|194|3709900622|4294967295|171177928|4294967295|4294967295|4294967295|4294967295|4294967295|42949672954294967295|1682962441|4294967295|4294967295|4294967295|1682942053|36412|36412|29101|77549443|65535|4294967295|cmnet.mnc002.mcc460.gprs|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|0|18327663707|460025071280390|3594540725330202|35945407|712|1501378581|1501378581|65535|18446744073709551615|4294967295|255|255|109|118|255";

	private static void createDir(String dir) throws IOException {
		String path = url + "/" + dir;
		if (!storage.exists(path)) {
			storage.mkdirs(path);
		}
	}

	public static void main(String[] args) throws Exception {

		String topicN = "LTES1MMEIN";
		url = "hdfs://hbbdcs-nn-01:8020";
		//url = "hdfs://10.1.235.32:8020";
		System.setProperty("HADOOP_USER_NAME", "hdfs");
		
		String topicsDir = "/for-kafka/topics";
		 

		if (args.length > 0) {
			if (args[0].length() > 1) {
				if (args[0].charAt(0) == '1') {
					onlyread = true;
				}
				if (args[0].charAt(1) == '1') {
					onlywrite = true;
				}
			}
		}
		System.out.println("onlyread  :"  + onlyread);
		System.out.println("onlywrite :"  + onlywrite);

		@SuppressWarnings("unchecked")
		Class<? extends Storage> storageClass = (Class<? extends Storage>) Class
				.forName("io.confluent.connect.hdfs.storage.HdfsStorage");
		storage = StorageFactory.createStorage(storageClass, new Configuration(), url);

		createDir(topicsDir);

		Path path = new Path(url + "/" + topicsDir + "/" + UUID.randomUUID().toString());
		final FSDataOutputStream out = path.getFileSystem(new Configuration()).create(path);
		final BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out));

		
		// consumer.seekToBeginning(consumer.assignment());

		if(onlywrite){
			while(true){
				writer.write(example+ lineSeparator); 
			}
		}
		
		KafkaConsumer<String, String> consumer = SplitSomeToTestTopic.getConsumer(topicN);
		while (true) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			for (ConsumerRecord<String, String> record : records) {
				if (onlyread)
					continue;
				writer.write(record.value().toString() + lineSeparator);
				// writer.flush();
			}
		}
	}
}
